// $Id: daemon.c 86 2011-05-04 17:33:43Z cedric.shih@gmail.com $
/*
 * Copyright 2011 Cedric Shih (cedric dot shih at gmail dot com)
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

#include <stdio.h>
#include <stdlib.h>
#include <string.h>

#include <nio.h>
#include <nio/buffer.h>
#include <nio/channel.h>
#include <nio/tcp.h>
#include <nio/dump.h>
#include <nio/server.h>
#include <niohttp.h>
#include <niohttp/server.h>

#include "log.h"

#include "stream.h"
#include "handler.h"
#include "dispatcher.h"
#include "version.h"

#include "daemon.h"

struct niohttpd *niohttpd_new(struct nioreactor *reactor)
{
	struct niohttpd *server;

	niohttpd_debug("constructing");

	if (!(server = calloc(1, sizeof(*server)))) {
		niohttpd_error("failed to calloc");
		goto exception;
	}

	if (!(server->server = nioserver_new(reactor,
			&niohttpd_channel_factory, server))) {
		niohttpd_error("failed to create server bootstrap: %s",
				strerror(errno));
		goto exception;
	}

	if (!(server->handlers = niohttpd_handlers_new())) {
		niohttpd_error("failed to create handler context: %s",
				strerror(errno));
		goto exception;
	}

	if (!(server->streams = niohttpd_streams_new())) {
		niohttpd_error("failed to create stream context: %s",
				strerror(errno));
		goto exception;
	}

	server->reactor = reactor;
	return server;

exception:
	niohttpd_free(server);
	return NULL;
}

void niohttpd_free(struct niohttpd *server)
{
	struct niohttpd_stream_entry *stream;
	struct niohttpd_handler_entry *handler;

	niohttpd_debug("destructing: %p", server);

	if (!server) {
		return;
	}

	if (server->streams) {
		while ((stream = RB_MIN(niohttpd_stream_tree,
				&server->streams->tree))) {
			niohttpd_remove_stream(server, stream);
		}

		niohttpd_streams_free(server->streams);
	}

	if (server->handlers) {
		while ((handler = RB_MIN(niohttpd_handler_tree,
				&server->handlers->tree))) {
			niohttpd_remove_handler(server, handler);
		}

		niohttpd_handlers_free(server->handlers);
	}

	if (server->server) {
		nioserver_free(server->server);
	}

	free(server);
}

inline int niohttpd_bind(struct niohttpd *server,
		struct in_addr *address, unsigned short port, int backlog)
{
	return nioserver_bind(server->server, address, port, backlog);
}

struct niobuffer *niohttpd_buffer_create(
		struct niochannel *channel, void *arg)
{
	struct niobuffer *buffer = NULL;

	if (!(buffer = niobuffer_new(1048576))) {
		niohttpd_error("failed to create buffer");
		goto exception;
	}

	return buffer;

exception:
	if (buffer) {
		niobuffer_free(buffer);
	}
	return NULL;
}

void niohttpd_buffer_release(struct niobuffer *buffer, void *arg)
{
	niobuffer_free(buffer);
}

struct niochannel *niohttpd_channel_create(
		struct niosocket *socket, void *arg)
{
	int rc;
	struct niohttpd *server = arg;
	struct niotcp_handler *tcp = NULL;
	struct niodump_handler *dump = NULL;
	struct niohttpd_dispatcher *dispatcher = NULL;
	struct niochannel *channel = NULL;

	if (!(channel = niochannel_new(server->reactor,
			&niohttpd_buffer_factory, NULL))) {
		niohttpd_error("failed to create channel");
		goto exception;
	}

	if (!(tcp = niotcp_handler_new(server->reactor, NULL))) {
		rc = errno ? errno : -1;
		niohttpd_error("failed to create tcp handler");
		goto exception;
	}

	if ((rc = niochannel_add_last(channel, "tcp",
			&niotcp_handler, tcp))) {
		niohttpd_error("failed to add tcp handler: %s",
				nio_strerror(rc));
		goto exception;
	}

	if (server->dump) {
		if (!(dump = niodump_handler_new(stdout))) {
			rc = errno ? errno : -1;
			niohttpd_error("failed to create dump handler");
			goto exception;
		}

		if ((rc = niochannel_add_last(channel, "dump",
				&niodump_handler, dump))) {
			niohttpd_error("failed to add dump handler: %s",
					nio_strerror(rc));
			goto exception;
		}
	}

	if (!(dispatcher = niohttpd_dispatcher_new(server))) {
		rc = errno ? errno : -1;
		niohttpd_error("failed to create server dispatcher");
		goto exception;
	}

	if ((rc = niochannel_add_last(channel, niohttpd_get_name(),
			&niohttpd_dispatcher, dispatcher))) {
		niohttpd_error("failed to add handler dispatcher: %s",
				nio_strerror(rc));
		goto exception;
	}

	return channel;

exception:
	if (channel) {
		niochannel_free(channel);
	}
	if (dispatcher) {
		niohttpd_dispatcher_free(dispatcher);
	}
	if (dump) {
		niodump_handler_free(dump);
	}
	if (tcp) {
		niotcp_handler_free(tcp);
	}
	return NULL;
}

void niohttpd_channel_release(struct niochannel *channel, void *arg)
{
	if (channel) {
		niochannel_free(channel);
	}
}

struct niobuffer_factory niohttpd_buffer_factory = {
		niohttpd_buffer_create,
		niohttpd_buffer_release,
};

struct niochannel_factory niohttpd_channel_factory = {
		niohttpd_channel_create,
		niohttpd_channel_release,
};

int niohttpd_add_handler(struct niohttpd *server, const char *uri,
		struct niohttpd_handler_factory *factory, void *arg)
{
	int rc;
	struct niohttpd_handler_entry *handler = NULL;

	if (!server || !uri || !factory) {
		niohttpd_error("null argument(s)");
		rc = NIOHTTPD_EINVAL;
		goto finally;
	}

	niohttpd_debug("adding handler: %s", uri);

	if ((rc = niohttpd_handlers_add(server->handlers, uri, &handler))) {
		niohttpd_error("failed to add handler '%s': %s", uri, strerror(rc));
		goto finally;
	}

	memcpy(&handler->factory, factory, sizeof(handler->factory));
	handler->cbarg = arg;
	rc = 0;

finally:
	if (rc != 0) {
		if (handler) {
			niohttpd_handlers_remove(server->handlers, handler);
		}
	}
	return rc;
}

int niohttpd_del_handler(struct niohttpd *server, const char *uri)
{
	int rc;
	struct niohttpd_handler_entry *handler;

	if (!server || !uri) {
		niohttpd_error("null argument(s)");
		rc = NIOHTTPD_EINVAL;
		goto finally;
	}

	niohttpd_debug("removing handler: %s", uri);

	if (!(handler = niohttpd_handlers_find(server->handlers, uri))) {
		rc = ENOENT;
		niohttpd_error("no such handler: %s", uri);
		goto finally;
	}

	niohttpd_remove_handler(server, handler);
	rc = 0;

finally:
	return rc;
}

inline void niohttpd_remove_handler(struct niohttpd *server,
		struct niohttpd_handler_entry *handler)
{
	niohttpd_handlers_remove(server->handlers, handler);
}

inline struct niohttpd_handler_entry *niohttpd_match_handler(
		struct niohttpd *server, const char *uri)
{
	return niohttpd_handlers_find(server->handlers, uri);
}

int niohttpd_add_stream(struct niohttpd *server, const char *uri,
		struct niohttp_headers *headers,
		niohttpd_stream_cb cb, void *arg)
{
	int rc;
	struct niohttpd_stream_entry *stream = NULL;

	niohttpd_debug("adding stream: %s", uri);

	if ((rc = niohttpd_streams_add(server->streams, uri, &stream))) {
		niohttpd_error("failed to add stream '%s': %s", uri, strerror(rc));
		goto finally;
	}

	if (headers && !(stream->headers = niohttp_headers_clone(
			headers, 1))) {
		rc = errno ? errno : -1;
		niohttpd_error("failed to clone headers: %s", strerror(rc));
		goto finally;
	}

	stream->cb = cb;
	stream->cbarg = arg;
	rc = 0;

finally:
	if (rc != 0) {
		if (stream) {
			niohttpd_streams_remove(server->streams, stream);
		}
	}
	return rc;
}

int niohttpd_del_stream(struct niohttpd *server, const char *uri)
{
	int rc;
	struct niohttpd_stream_entry *stream;

	niohttpd_debug("removing stream: %s", uri);

	if (!(stream = niohttpd_streams_find(server->streams, uri))) {
		rc = ENOENT;
		niohttpd_error("no such stream: %s", uri);
		goto finally;
	}

	niohttpd_remove_stream(server, stream);
	rc = 0;

finally:
	return rc;
}

void niohttpd_on_stream_client_writable(struct niochannel *channel, void *arg)
{
	int rc;
	struct niohttpd_client *client = arg;
	struct niohttpd_stream_message *message;

	if (!(message = TAILQ_FIRST(&client->messages))) {
		niohttpd_info("writing multipart message");
		return;
	}

	niohttpd_info("writing multipart message");

	if ((rc = niochannel_write(channel, &message->multipart))) {
		niohttpd_error("failed to write multipart message: %s",
				strerror(rc));
	} else {
		TAILQ_REMOVE(&client->messages, message, next);
	}
}

int niohttpd_add_stream_client(struct niohttpd *server, const char *uri,
		struct niochannel *channel)
{
	int rc;
	struct niohttpd_stream_entry *stream;
	struct niohttpd_stream_event event;
	struct niohttpd_client *client = NULL;

	niohttpd_debug("adding client to: %s", uri);

	if (!(stream = niohttpd_streams_find(server->streams, uri))) {
		rc = ENOENT;
		niohttpd_error("no such stream: %s", uri);
		goto finally;
	}

	niohttpd_debug("adding connection as client");

	if (!(client = niohttpd_clients_add(&stream->clients,
			stream, channel))) {
		rc = errno ? errno : -1;
		niohttpd_error("failed to add client");
		goto finally;
	}

	if ((rc = niochannel_on_writable(channel,
			niohttpd_on_stream_client_writable, client))) {
		niohttpd_error("failed to listen channel writable: %s",
				nio_strerror(rc));
		goto finally;
	}

	if (stream->cb) {
		memset(&event, '\0', sizeof(event));
		event.type = NIOHTTPD_STREAM_EVENT_OPEN;

		niohttpd_debug("invoking cb for open: %p", stream->cb);

		(*stream->cb)(&event, stream->cbarg);
	}

	rc = 0;

finally:
	if (rc != 0) {
		if (client) {
			niohttpd_remove_stream_client(server, client, 0);
		}
	}
	return rc;
}

int niohttpd_close_stream(struct niohttpd *server, const char *uri)
{
	int rc;
	struct niohttpd_stream_entry *stream;
	struct niohttpd_client *client;

	niohttpd_debug("closing all connections: %s", uri);

	if (!(stream = niohttpd_streams_find(server->streams, uri))) {
		rc = ENOENT;
		niohttpd_error("no such stream: %s", uri);
		goto finally;
	}

	while ((client = TAILQ_FIRST(&stream->clients))) {
		niohttpd_remove_stream_client(server, client, 1);
	}

	rc = 0;

finally:
	return rc;
}

int niohttpd_set_stream_qos(struct niohttpd *server, const char *uri,
		enum niohttpd_stream_qos_direction dir,
		niohttpd_stream_qos_cb cb, void *cbarg)
{
	int rc;
	struct niohttpd_stream_entry *stream;

	niohttpd_debug("setting QoS on stream: %s", uri);

	if (!(stream = niohttpd_streams_find(server->streams, uri))) {
		rc = ENOENT;
		niohttpd_error("no such stream: %s", uri);
		goto finally;
	}

	stream->qos.dir = dir;
	stream->qos.cb = cb;
	stream->qos.cbarg = cbarg;
	rc = 0;

finally:
	return rc;
}

void niohttpd_remove_stream(struct niohttpd *server,
		struct niohttpd_stream_entry *stream)
{
	struct niohttpd_client *client;

	while ((client = TAILQ_FIRST(&stream->clients))) {
		niohttpd_remove_stream_client(server, client, 1);
	}

	niohttpd_streams_remove(server->streams, stream);
}

struct niohttpd_stream_message *niohttpd_stream_qos_candidate(
		struct niohttpd_client *client)
{
	struct niohttpd_stream_message *candidate = NULL;

	niohttpd_trace("finding candidate message to discard");

	switch (client->stream->qos.dir) {
	case NIOHTTPD_STREAM_QOS_FROM_OLDEST:
		niohttpd_trace("trying first message");
		candidate = TAILQ_FIRST(&client->messages);
//
//		if (candidate && candidate->position != 0) {
//			niohttpd_trace("transmission in progress; trying next");
//			candidate = TAILQ_NEXT(candidate, next);
//		}
		break;
	case NIOHTTPD_STREAM_QOS_FROM_LATEST:
		niohttpd_trace("trying last message");
		candidate = TAILQ_LAST(&client->messages,
				niohttpd_stream_messages);
//
//		if (candidate && candidate->position != 0) {
//			niohttpd_trace("transmission in progress");
//			candidate = NULL;
//		}
		break;
	}

	return candidate;
}

int niohttpd_add_stream_message(struct niohttpd_stream_entry *stream,
		struct niohttpd_client *client,
		struct niohttp_headers *headers, struct niobuffer *content)
{
	int rc;
	struct niohttpd_stream_message *candidate, *message = NULL;
	struct niohttpd_stream_context ctx;
	struct niohttp_headers *discard;

	if (!TAILQ_EMPTY(&client->messages)) {
		if (!stream->qos.cb) {
			niohttpd_trace("skipping incoming message due to no "
					"QoS");
			rc = 0;
			goto finally;
		}

		niohttpd_trace("starting QoS analysis");

		while ((candidate = niohttpd_stream_qos_candidate(
				client))) {
			memset(&ctx, '\0', sizeof(ctx));
			ctx.queued = candidate->multipart.headers;
			ctx.incoming = headers;

			niohttpd_trace("invoking QoS callback: %p",
					stream->qos.cb);

			if (!(discard = (*stream->qos.cb)(stream->uri,
					stream->qos.dir,
					&ctx, stream->qos.cbarg))) {
				niohttpd_trace("don't discard any message");
				break;
			}

			if (discard == candidate->multipart.headers) {
				niohttpd_trace("discarding previous message");

				niohttpd_stream_messages_remove(
						&client->messages,
						candidate);
			} else {
				if (discard != headers) {
					niohttpd_warn("invalid QoS result");
				}

				niohttpd_trace("discarding incoming message");
				rc = 0;
				goto finally;
			}
		}
	}

	if (!(message = niohttpd_stream_messages_add(
			&client->messages))) {
		niohttpd_error("failed to add message");
		rc = ENOMEM;
		goto finally;
	}

	message->multipart.headers = headers;
	message->multipart.content = content;

	niohttpd_on_stream_client_writable(client->channel, client);
	rc = 0;

finally:
	if (rc != 0) {
		if (message) {
			niohttpd_stream_messages_remove(
					&client->messages, message);
		}
	}
	return rc;
}

int niohttpd_push_stream(struct niohttpd *server, const char *uri,
		struct niohttp_headers *headers, struct niobuffer *content)
{
	int rc;
	struct niohttpd_stream_entry *stream;
	struct niohttpd_client *client;

	niohttpd_info("pushing: %s", uri);

	if (!(stream = niohttpd_streams_find(server->streams, uri))) {
		rc = ENOENT;
		niohttpd_error("no such stream: %s", uri);
		goto finally;
	}

	TAILQ_FOREACH(client , &stream->clients, next) {
		if (client->error) {
			niohttpd_debug("skipping error client: %p", client);
			continue;
		}

		niohttpd_info("pushing message to client: %p", client);

		if ((rc = niohttpd_add_stream_message(stream, client,
				headers, content))) {

			niohttpd_error("failed to add message: %s",
					strerror(rc));
			client->error = 1;
		}
	}

	rc = 0;

finally:
	return rc;
}

inline struct niohttpd_stream_entry *niohttpd_match_stream(
		struct niohttpd *server, const char *uri)
{
	return niohttpd_streams_find(server->streams, uri);
}

void niohttpd_remove_stream_client(struct niohttpd *server,
		struct niohttpd_client *client, int notify)
{
	struct niohttpd_stream_event event;

	niohttpd_debug("closing client: %p", client);

	if (notify && client->stream->cb) {
		memset(&event, '\0', sizeof(event));
		event.type = NIOHTTPD_STREAM_EVENT_CLOSING;

		niohttpd_debug("invoking cb for closing: %p",
				client->stream->cb);

		(*client->stream->cb)(&event, client->stream->cbarg);
	}

	niohttpd_debug("remove from list");

	niohttpd_clients_remove(&client->stream->clients, client);
}
